/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.hadoop.hdfs.server.namenode;

import static org.apache.hadoop.util.Time.now;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.NavigableSet;
import java.util.NoSuchElementException;
import java.util.SortedMap;
import java.util.TreeMap;
import java.util.TreeSet;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.UnresolvedLinkException;
import org.apache.hadoop.hdfs.protocol.HdfsConstants;
import org.apache.hadoop.hdfs.server.blockmanagement.BlockInfo;
import org.apache.hadoop.hdfs.server.common.HdfsServerConstants;
import org.apache.hadoop.util.Daemon;

import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;

/**
 * LeaseManager does the lease housekeeping for writing on files.   
 * This class also provides useful static methods for lease recovery.
 * 
 * Lease Recovery Algorithm
 * 1) Namenode retrieves lease information
 * 2) For each file f in the lease, consider the last block b of f
 * 2.1) Get the datanodes which contains b
 * 2.2) Assign one of the datanodes as the primary datanode p

 * 2.3) p obtains a new generation stamp from the namenode
 * 2.4) p gets the block info from each datanode
 * 2.5) p computes the minimum block length
 * 2.6) p updates the datanodes, which have a valid generation stamp,
 *      with the new generation stamp and the minimum block length 
 * 2.7) p acknowledges the namenode the update results

 * 2.8) Namenode updates the BlockInfo
 * 2.9) Namenode removes f from the lease
 *      and removes the lease once all files have been removed
 * 2.10) Namenode commit changes to edit log
 */
@InterfaceAudience.Private
public class LeaseManager {
  public static final Log LOG = LogFactory.getLog(LeaseManager.class);

  private final FSNamesystem fsnamesystem;

  private long softLimit = HdfsConstants.LEASE_SOFTLIMIT_PERIOD;
  private long hardLimit = HdfsConstants.LEASE_HARDLIMIT_PERIOD;

  //
  // Used for handling lock-leases
  // Mapping: leaseHolder -> Lease
  //
  /**
   * TreeMap
   * 如果大家去关注一下我之前讲解的那个 JDK 源码剖析系列的课程
   * 里面就有集合包的源码剖析
   * 里面是提到了这个 TreeMap 的，红黑树保存的集合数据结构，特点就是说放入里面的数据
   * 可以默认自动根据 key 的自然大小来排序
   * 排序好了以后，下次你如果遍历 TreeMap 的时候，就是根据 key 自然大小排序后的顺序来遍历
   *
   * HashMap 遍历的时候是无序的
   * LinkedHashMap 是用一个双向链表保存了插入 map 的 key-value 对的顺序
   * 遍历 LinkedHashMap 的时候是根据插入的顺序来遍历的
   * TreeMap 是默认根据 key 的自然大小来排序的，遍历的时候也是按照这个顺序来遍历的
   *
   * 用 TreeMap 保存了一个 lease 数据集合
   * 想想就知道，lease 一看就是契约，是 leaseManager 管理的核心数据
   * 一个 lease 就是代表了某个客户端对一个文件拥有的契约
   * leases数据结构里，key 是客户端名称，value 就是那个客户端的 leases 契约
   * TreeMap 的话，默认会根据客户端的名称来进行排序
   */
  private final SortedMap<String, Lease> leases = new TreeMap<String, Lease>();
  // Set of: Lease
  /**
   * TreeSet 如果看过 JDK 源码剖析的课程
   * set 系列的东西是没有源码的，他全部是基于 map 系列的类来实现的
   * TreeSet 底层就是基于 TreeMap 来实现的，默认里面的元素就是按照自然大小排序的
   * 普通的 hashSet 是无序的
   * 相当于默认就是根据 lease 来自然排序了
   * 默认是根据 lease 的 lastUpdate 续约时间来排序，如果 lastUpdate 一样，那么就根据客户端来排序
   */
  private final NavigableSet<Lease> sortedLeases = new TreeSet<Lease>();

  // 
  // Map path names to leases. It is protected by the sortedLeases lock.
  // The map stores pathnames in lexicographical order.
  //
  /**
   * TreeSet 根据 path 来排序的契约数据
   * key 是文件路径名，默认是根据文件路径名来进行排序的
   */
  private final SortedMap<String, Lease> sortedLeasesByPath = new TreeMap<String, Lease>();

  /**
   * 核心：Daemon，本身就是基于他创建的线程，都是 Daemon 后台线程
   */
  private Daemon lmthread;
  /**
   * 有一个东西，是叫做 shouldRunMonitor 标志位
   * volatile 修饰的标志位，volatile 是什么，请看 java 并发编程
   * 有详细的 volatile 和 java 内存模型的原理
   */
  private volatile boolean shouldRunMonitor;

  LeaseManager(FSNamesystem fsnamesystem) {this.fsnamesystem = fsnamesystem;}

  Lease getLease(String holder) {
    return leases.get(holder);
  }

  @VisibleForTesting
  int getNumSortedLeases() {return sortedLeases.size();}

  /**
   * This method iterates through all the leases and counts the number of blocks
   * which are not COMPLETE. The FSNamesystem read lock MUST be held before
   * calling this method.
   *
   * 这个东西大概的意思，是获取正在构造中的 block(under construction 状态)
   * 是在我们最早最早分析的那个 nameNode 启动的时候会进行一个 safe mode 的检查，当时会调用这个方法
   * @return
   */
  synchronized long getNumUnderConstructionBlocks() {
    assert this.fsnamesystem.hasReadLock() : "The FSNamesystem read lock wasn't"
      + "acquired before counting under construction blocks";
    long numUCBlocks = 0;
    /**
     * TODO 这里就是遍历所有的 lease ，lease 契约的东西，这个后面详细讲解
     * 每个 lease 里面有多个 path ，你可以认为一个 path 就是代表了一个文件的路径，/usr/warehouse/hive/access.log
     */
    for (Lease lease : sortedLeases) {
      for (String path : lease.getPaths()) {
        final INodeFile cons;
        try {
          /**
           * 会通过 FSNameSystem 里面有一个关键的东西，FSDirectory().getINode(path).asFile()
           * 就可以获取到代表了 /usr/warehouse/hive/access.log 这个东西的一个文件对象 INodeFile
           */
          cons = this.fsnamesystem.getFSDirectory().getINode(path).asFile();
          /**
           * 这里会判断一下，这个文件当前是否处于 under construction 状态，如果不是处于这个状态，那么就直接看下一个 path
           */
          if (!cons.isUnderConstruction()) {
            LOG.warn("The file " + cons.getFullPathName()
                + " is not under construction but has lease.");
            continue;
          }
        } catch (UnresolvedLinkException e) {
          throw new AssertionError("Lease files should reside on this FS");
        }
        /**
         * 这边是通过 INodeFile.getBlocks() 的方法
         * 获取 /usr/warehouse/hive/access.log 文件的 block
         * 每个文件都可以对应多个 block，可以拆分为128m一个的 block，所以这里一个文件可能有多个 block
         */
        BlockInfo[] blocks = cons.getBlocks();
        if(blocks == null)
          continue;

        /**
         * 遍历这个文件的 block
         */
        for(BlockInfo b : blocks) {
          /**
           * 如果 BlockInfo.isComplete()方法判断当前这个 block 他的副本数量不足够3个，就会标识为 complete = false
           * update: 如果这个 block 的状态不是 complete ，那么这个 block 就属于一个 under construction 的 block
           */
          if(!b.isComplete())
            numUCBlocks++;
        }
      }
    }
    LOG.info("Number of blocks under construction: " + numUCBlocks);
    return numUCBlocks;
  }

  /** @return the lease containing src */
  public Lease getLeaseByPath(String src) {return sortedLeasesByPath.get(src);}

  /** @return the number of leases currently in the system */
  public synchronized int countLease() {return sortedLeases.size();}

  /** @return the number of paths contained in all leases */
  synchronized int countPath() {
    int count = 0;
    for(Lease lease : sortedLeases) {
      count += lease.getPaths().size();
    }
    return count;
  }
  
  /**
   * Adds (or re-adds) the lease for the specified file.
   *
   * 为某个客户端加入一个针对某个文件的契约
   */
  synchronized Lease addLease(String holder, String src) {
    /**
     * holder 一看就是代表客户端的名称
     * 当然刚开始肯定是找不到的
     */
    Lease lease = getLease(holder);
    if (lease == null) {
      /**
       * 直接构造一个内部类，Lease 对象，传入进去客户端的名称
       */
      lease = new Lease(holder);
      /**
       * 直接以<客户端名称，Lease契约> 的 key-value 对格式，将他放入 Lease 数据结构中
       */
      leases.put(holder, lease);
      /**
       * 同时将 Lease 数据加入了一个 sortedLeases 的数据结构中
       */
      sortedLeases.add(lease);
    } else {
      renewLease(lease);
    }
    /**
     * 往 sortedLeasesByPath 的数据结构中加入契约
     * 这个数据结构的 key，不是客户端的名称，而是文件的路径名。所以他默认是根据文件路径名来排序的
     */
    sortedLeasesByPath.put(src, lease);
    lease.paths.add(src);
    return lease;
  }

  /**
   * Remove the specified lease and src.
   * 删除某个客户端针对某个文件的契约
   * 从他所有的数据结构中，将这个文件契约给删除掉
   */
  synchronized void removeLease(Lease lease, String src) {
    sortedLeasesByPath.remove(src);
    if (!lease.removePath(src)) {
      if (LOG.isDebugEnabled()) {
        LOG.debug(src + " not found in lease.paths (=" + lease.paths + ")");
      }
    }

    if (!lease.hasPath()) {
      leases.remove(lease.holder);
      if (!sortedLeases.remove(lease)) {
        LOG.error(lease + " not found in sortedLeases");
      }
    }
  }

  /**
   * Remove the lease for the specified holder and src
   */
  synchronized void removeLease(String holder, String src) {
    Lease lease = getLease(holder);
    if (lease != null) {
      removeLease(lease, src);
    } else {
      LOG.warn("Removing non-existent lease! holder=" + holder +
          " src=" + src);
    }
  }

  synchronized void removeAllLeases() {
    sortedLeases.clear();
    sortedLeasesByPath.clear();
    leases.clear();
  }

  /**
   * Reassign lease for file src to the new holder.
   */
  synchronized Lease reassignLease(Lease lease, String src, String newHolder) {
    assert newHolder != null : "new lease holder is null";
    if (lease != null) {
      removeLease(lease, src);
    }
    return addLease(newHolder, src);
  }

  /**
   * Renew the lease(s) held by the given client
   *
   * Renew，如果大家看过我之前分析的 spring cloud 源码的课的话，就知道里面
   * 微服务注册中心，eureka 里面有 lease 和 renew 的概念
   * 各个服务都会不断的发送心跳给 eureka server，里面会维护每个服务跟 eureka server 之间的 lease
   * 每发送一次细条，就会 renew lease 续约
   */
  synchronized void renewLease(String holder) {
    renewLease(getLease(holder));
  }
  synchronized void renewLease(Lease lease) {
    if (lease != null) {
      /**
       * 先从 sortedLeases 里面删除掉，因为他要重新续约
       * 更新最近一次续约的时间
       * 重新加入 sortedLeases 里面去，根据最新的续约时间重新排序
       */
      sortedLeases.remove(lease);
      lease.renew();
      sortedLeases.add(lease);
    }
  }

  /**
   * Renew all of the currently open leases.
   */
  synchronized void renewAllLeases() {
    for (Lease l : leases.values()) {
      renewLease(l);
    }
  }

  /************************************************************
   * A Lease governs all the locks held by a single client.
   * For each client there's a corresponding lease, whose
   * timestamp is updated when the client periodically
   * checks in.  If the client dies and allows its lease to
   * expire, all the corresponding locks can be released.
   *************************************************************/
  class Lease implements Comparable<Lease> {
    /**
     * holder 哪个客户端持有的这份契约
     */
    private final String holder;
    /**
     * 客户端，一看就是说，肯定是有了一份契约之后，必须得用一个后台线程不断发送请求，来进行 renew lease 续约
     * 这个 lastUpdate 一般存储的是他最近一次进行续约的时间
     */
    private long lastUpdate;
    /**
     * 这个客户端在这份契约里针对那些文件声明了自己的所有权
     */
    private final Collection<String> paths = new TreeSet<String>();
  
    /** Only LeaseManager object can create a lease */
    private Lease(String holder) {
      this.holder = holder;
      renew();
    }
    /** Only LeaseManager object can renew a lease */
    private void renew() {
      this.lastUpdate = now();
    }

    /** @return true if the Hard Limit Timer has expired */
    public boolean expiredHardLimit() {
      /**
       * 当前时间 - 最后更新时间 > 1h
       */
      return now() - lastUpdate > hardLimit;
    }

    /** @return true if the Soft Limit Timer has expired */
    public boolean expiredSoftLimit() {
      return now() - lastUpdate > softLimit;
    }

    /** Does this lease contain any path? */
    boolean hasPath() {return !paths.isEmpty();}

    boolean removePath(String src) {
      return paths.remove(src);
    }

    @Override
    public String toString() {
      return "[Lease.  Holder: " + holder
          + ", pendingcreates: " + paths.size() + "]";
    }
  
    @Override
    public int compareTo(Lease o) {
      Lease l1 = this;
      Lease l2 = o;
      /**
       * 研究一下，各个 Lease 是根据什么来排序的
       */
      long lu1 = l1.lastUpdate;
      long lu2 = l2.lastUpdate;
      /**
       * 是根据 lastUpdate 也就是最近一次续约的时间来进行排序的
       */
      if (lu1 < lu2) {
        return -1;
      } else if (lu1 > lu2) {
        return 1;
      } else {
        /**
         * 如果两个客户端的续约时间是一样的，那么就根据客户端的名称来排序
         */
        return l1.holder.compareTo(l2.holder);
      }
    }
  
    @Override
    public boolean equals(Object o) {
      if (!(o instanceof Lease)) {
        return false;
      }
      Lease obj = (Lease) o;
      if (lastUpdate == obj.lastUpdate &&
          holder.equals(obj.holder)) {
        return true;
      }
      return false;
    }
  
    @Override
    public int hashCode() {
      return holder.hashCode();
    }
    
    Collection<String> getPaths() {
      return paths;
    }

    String getHolder() {
      return holder;
    }

    void replacePath(String oldpath, String newpath) {
      paths.remove(oldpath);
      paths.add(newpath);
    }
    
    @VisibleForTesting
    long getLastUpdate() {
      return lastUpdate;
    }
  }

  synchronized void changeLease(String src, String dst) {
    if (LOG.isDebugEnabled()) {
      LOG.debug(getClass().getSimpleName() + ".changelease: " +
               " src=" + src + ", dest=" + dst);
    }

    final int len = src.length();
    for(Map.Entry<String, Lease> entry
        : findLeaseWithPrefixPath(src, sortedLeasesByPath).entrySet()) {
      final String oldpath = entry.getKey();
      final Lease lease = entry.getValue();
      // replace stem of src with new destination
      final String newpath = dst + oldpath.substring(len);
      if (LOG.isDebugEnabled()) {
        LOG.debug("changeLease: replacing " + oldpath + " with " + newpath);
      }
      lease.replacePath(oldpath, newpath);
      sortedLeasesByPath.remove(oldpath);
      sortedLeasesByPath.put(newpath, lease);
    }
  }

  synchronized void removeLeaseWithPrefixPath(String prefix) {
    for(Map.Entry<String, Lease> entry
        : findLeaseWithPrefixPath(prefix, sortedLeasesByPath).entrySet()) {
      if (LOG.isDebugEnabled()) {
        LOG.debug(LeaseManager.class.getSimpleName()
            + ".removeLeaseWithPrefixPath: entry=" + entry);
      }
      removeLease(entry.getValue(), entry.getKey());    
    }
  }

  static private Map<String, Lease> findLeaseWithPrefixPath(
      String prefix, SortedMap<String, Lease> path2lease) {
    if (LOG.isDebugEnabled()) {
      LOG.debug(LeaseManager.class.getSimpleName() + ".findLease: prefix=" + prefix);
    }

    final Map<String, Lease> entries = new HashMap<String, Lease>();
    int srclen = prefix.length();
    
    // prefix may ended with '/'
    if (prefix.charAt(srclen - 1) == Path.SEPARATOR_CHAR) {
      srclen -= 1;
    }

    for(Map.Entry<String, Lease> entry : path2lease.tailMap(prefix).entrySet()) {
      final String p = entry.getKey();
      if (!p.startsWith(prefix)) {
        return entries;
      }
      if (p.length() == srclen || p.charAt(srclen) == Path.SEPARATOR_CHAR) {
        entries.put(entry.getKey(), entry.getValue());
      }
    }
    return entries;
  }

  public void setLeasePeriod(long softLimit, long hardLimit) {
    this.softLimit = softLimit;
    this.hardLimit = hardLimit; 
  }
  
  /******************************************************
   * Monitor checks for leases that have expired,
   * and disposes of them.
   * 这是个后台线程
   * 这个东西默认一定会运行起来，在后台不断地额监控各个 lease 契约
   * 如果某个客户端申请持有了一个文件的契约
   * 结果，那个客户端莫名其妙的死亡了，然后没有删除这个契约
   * 那么此时就需要 nameNode 里，LeaseManager 的 Monitor 后台线程
   * 不断的监控各个 lease，判断如果说某个客户端申请了契约之后，长时间范围内都没有来续约，那么就认为他死了
   * 此时就要自动释放这个 lease，不让一个客户端长时间占有这个文件的所有权
   ******************************************************/
  class Monitor implements Runnable {
    final String name = getClass().getSimpleName();

    /** Check leases periodically. */
    @Override
    public void run() {
      for(; shouldRunMonitor && fsnamesystem.isRunning(); ) {
        boolean needSync = false;
        try {
          fsnamesystem.writeLockInterruptibly();
          try {
            if (!fsnamesystem.isInSafeMode()) {
              /**
               * 检查契约的核心逻辑，是在 checkLeases() 方法里
               * 在这个检查契约的逻辑中，可能会触发一些写 edits log 元数据的操作
               * 如果触发了的话，还要让 edits log sync 一下
               */
              needSync = checkLeases();
            }
          } finally {
            fsnamesystem.writeUnlock();
            // lease reassignments should to be sync'ed.
            if (needSync) {
              fsnamesystem.getEditLog().logSync();
            }
          }

          /**
           * 默认是每隔 2s，对所有的契约做一次检查
           */
          Thread.sleep(HdfsServerConstants.NAMENODE_LEASE_RECHECK_INTERVAL);
        } catch(InterruptedException ie) {
          if (LOG.isDebugEnabled()) {
            LOG.debug(name + " is interrupted", ie);
          }
        }
      }
    }
  }

  /**
   * Get the list of inodes corresponding to valid leases.
   * @return list of inodes
   */
  Map<String, INodeFile> getINodesUnderConstruction() {
    Map<String, INodeFile> inodes = new TreeMap<String, INodeFile>();
    for (String p : sortedLeasesByPath.keySet()) {
      // verify that path exists in namespace
      try {
        INodeFile node = INodeFile.valueOf(fsnamesystem.dir.getINode(p), p);
        if (node.isUnderConstruction()) {
          inodes.put(p, node);
        } else {
          LOG.warn("Ignore the lease of file " + p
              + " for checkpoint since the file is not under construction");
        }
      } catch (IOException ioe) {
        LOG.error(ioe);
      }
    }
    return inodes;
  }
  
  /** Check the leases beginning from the oldest.
   *
   * sortedLeases 默认是根据 lease 续约时间来进行排序的，默认是续约时间越旧，越老的，越靠前
   * 是放在第一个
   * 所以在这里可以 first() 方法获取第一个契约
   *
   *  @return true is sync is needed.
   */
  @VisibleForTesting
  synchronized boolean checkLeases() {
    boolean needSync = false;
    assert fsnamesystem.hasWriteLock();
    Lease leaseToCheck = null;
    try {
      leaseToCheck = sortedLeases.first();
    } catch(NoSuchElementException e) {}

    while(leaseToCheck != null) {
      /**
       * 如果最老的一个契约
       * 所有的契约都有一个续约的时间
       * 比如说：契约1：10:00:00，契约2：10:05:00，契约3：10:20:00
       * 人家在这里做检查，检查契约过期的时候，不是很僵硬的遍历所有的契约来检查他的时间，那样的话性能不是太好
       * 他就根据一个 sortedLeases 排序的数据结构，取出来续约时间最老的那个
       * 比如说 10:00:00
       * 如果说最老的这个契约，都没有超过一小时还没续约，那么就不用检查了，因为就意味着后面的契约肯定续约时间都没有超过一个小时
       */
      if (!leaseToCheck.expiredHardLimit()) {
        break;
      }

      /**
       * 如果进入到这里，发现有某个契约，已经超过了 1h 还没续约了
       */
      LOG.info(leaseToCheck + " has expired hard limit");

      final List<String> removing = new ArrayList<String>();
      // need to create a copy of the oldest lease paths, because 
      // internalReleaseLease() removes paths corresponding to empty files,
      // i.e. it needs to modify the collection being iterated over
      // causing ConcurrentModificationException
      String[] leasePaths = new String[leaseToCheck.getPaths().size()];
      leaseToCheck.getPaths().toArray(leasePaths);
      /**
       * 遍历这个契约中有权限的文件路径
       */
      for(String p : leasePaths) {
        try {
          /**
           * 此时就会自动释放掉那些契约
           * 释放掉一个契约针对某个文件路径的所有权
           */
          boolean completed = fsnamesystem.internalReleaseLease(leaseToCheck, p,
              HdfsServerConstants.NAMENODE_LEASE_HOLDER);
          if (LOG.isDebugEnabled()) {
            if (completed) {
              LOG.debug("Lease recovery for " + p + " is complete. File closed.");
            } else {
              LOG.debug("Started block recovery " + p + " lease " + leaseToCheck);
            }
          }
          // If a lease recovery happened, we need to sync later.
          if (!needSync && !completed) {
            needSync = true;
          }
        } catch (IOException e) {
          LOG.error("Cannot release the path " + p + " in the lease "
              + leaseToCheck, e);
          removing.add(p);
        }
      }

      for(String p : removing) {
        /**
         * 删除这个契约
         */
        removeLease(leaseToCheck, p);
      }
      /**
       * 获取到 sortedLeases 中的第二个契约，发现第一个契约是过期的
       */
      leaseToCheck = sortedLeases.higher(leaseToCheck);
    }

    try {
      if(leaseToCheck != sortedLeases.first()) {
        LOG.warn("Unable to release hard-limit expired lease: "
          + sortedLeases.first());
      }
    } catch(NoSuchElementException e) {}
    return needSync;
  }

  @Override
  public synchronized String toString() {
    return getClass().getSimpleName() + "= {"
        + "\n leases=" + leases
        + "\n sortedLeases=" + sortedLeases
        + "\n sortedLeasesByPath=" + sortedLeasesByPath
        + "\n}";
  }

  /**
   * 这个方法，一看就是在这里构造了这个后台线程
   * 然后同时启动了这个后台线程
   * 这个方法一定是在 FSNameSystem 初始化的时候来调用 LeaseManager 的方法，启动他的后台线程
   */
  void startMonitor() {
    Preconditions.checkState(lmthread == null,
        "Lease Monitor already running");
    shouldRunMonitor = true;
    lmthread = new Daemon(new Monitor());
    lmthread.start();
  }
  
  void stopMonitor() {
    if (lmthread != null) {
      shouldRunMonitor = false;
      try {
        lmthread.interrupt();
        lmthread.join(3000);
      } catch (InterruptedException ie) {
        LOG.warn("Encountered exception ", ie);
      }
      lmthread = null;
    }
  }

  /**
   * Trigger the currently-running Lease monitor to re-check
   * its leases immediately. This is for use by unit tests.
   */
  @VisibleForTesting
  void triggerMonitorCheckNow() {
    Preconditions.checkState(lmthread != null,
        "Lease monitor is not running");
    lmthread.interrupt();
  }
}
